// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#include <pollux/functions/sparksql/tests/SparkFunctionBaseTest.h>

namespace kumo::pollux::functions::sparksql::test {
namespace {

class DecimalCompareTest : public SparkFunctionBaseTest {
 protected:
  void testCompareExpr(
      const std::string& exprStr,
      const std::vector<VectorPtr>& input,
      const VectorPtr& expectedResult,
      const std::optional<SelectivityVector>& rows = std::nullopt) {
    auto actual = evaluate(exprStr, make_row_vector(input), rows);
    if (rows.has_value()) {
      pollux::test::assertEqualVectors(expectedResult, actual, rows.value());
    } else {
      pollux::test::assertEqualVectors(expectedResult, actual);
    }
  }
};

TEST_F(DecimalCompareTest, gt) {
  // Fast path when c1 vector is constant.
  testCompareExpr(
      "decimal_greaterthan(c0, c1)",
      {
          make_flat_vector<int64_t>({1000, 2000, 3000, 400}, DECIMAL(6, 2)),
          make_constant((int64_t)100, 4, DECIMAL(5, 1)),
      },
      make_flat_vector<bool>({false, true, true, false}));
  testCompareExpr(
      "decimal_greaterthan(c0, c1)",
      {
          make_flat_vector<int128_t>(
              {10'000'000'000'000'000,
               12'000'000'000'000'000,
               13'000'000'000'000'000,
               35'000'000'000'000'000},
              DECIMAL(38, 34)),
          make_constant((int64_t)100000, 4, DECIMAL(6, 1)),
      },
      make_flat_vector<bool>({false, false, false, false}));

  // Fast path when vectors are flat.
  testCompareExpr(
      "decimal_greaterthan(c0, c1)",
      {
          make_flat_vector<int64_t>({1000, 2000, 3000, 400}, DECIMAL(6, 2)),
          make_flat_vector<int64_t>({100, 120, 130, 350}, DECIMAL(5, 1)),
      },
      make_flat_vector<bool>({false, true, true, false}));
  testCompareExpr(
      "decimal_greaterthan(c0, c1)",
      {
          make_flat_vector<int128_t>(
              {10'000'000'000'000'000,
               12'000'000'000'000'000,
               13'000'000'000'000'000,
               35'000'000'000'000'000},
              DECIMAL(38, 34)),
          make_flat_vector<int64_t>(
              {100000, 120000, 130000, 350000}, DECIMAL(6, 1)),
      },
      make_flat_vector<bool>({false, false, false, false}));

  // General case when vectors are dictionary-encoded.
  testCompareExpr(
      "decimal_greaterthan(c0, c1)",
      {
          wrap_in_dictionary(
              make_indices({0, 1, 2, 3}),
              make_flat_vector<int64_t>({1000, 2000, 3000, 400}, DECIMAL(6, 2))),
          wrap_in_dictionary(
              make_indices({0, 1, 2, 3}),
              make_flat_vector<int64_t>({100, 120, 130, 350}, DECIMAL(5, 1))),
      },
      make_flat_vector<bool>({false, true, true, false}));
  testCompareExpr(
      "decimal_greaterthan(c0, c1)",
      {
          wrap_in_dictionary(
              make_indices({0, 1, 2, 3}),
              make_flat_vector<int128_t>(
                  {10'000'000'000'000'000,
                   12'000'000'000'000'000,
                   13'000'000'000'000'000,
                   35'000'000'000'000'000},
                  DECIMAL(38, 34))),
          wrap_in_dictionary(
              make_indices({0, 1, 2, 3}),
              make_flat_vector<int64_t>(
                  {100000, 120000, 130000, 350000}, DECIMAL(6, 1))),
      },
      make_flat_vector<bool>({false, false, false, false}));

  // Decimal with nulls.
  testCompareExpr(
      "decimal_greaterthan(c0, c1)",
      {
          make_flat_vector<int64_t>({1000, 2000, 3000, 400}, DECIMAL(6, 2)),
          make_nullable_flat_vector<int64_t>(
              {100, std::nullopt, 130, 350}, DECIMAL(5, 1)),
      },
      make_nullable_flat_vector<bool>({false, std::nullopt, true, false}));
  testCompareExpr(
      "decimal_greaterthan(c0, c1)",
      {
          make_flat_vector<int128_t>(
              {10'000'000'000'000'000,
               12'000'000'000'000'000,
               13'000'000'000'000'000,
               35'000'000'000'000'000},
              DECIMAL(38, 34)),
          make_nullable_flat_vector<int64_t>(
              {100000, std::nullopt, 130000, 350000}, DECIMAL(6, 1)),
      },
      make_nullable_flat_vector<bool>({false, std::nullopt, false, false}));

  // All rows selected.
  testCompareExpr(
      "decimal_greaterthan(c0, c1)",
      {
          make_flat_vector<int64_t>(
              70,
              [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
              nullptr,
              DECIMAL(6, 2)),
          make_flat_vector<int64_t>(
              70,
              [](auto row) { return row % 2 == 0 ? 100 : 130; },
              nullptr,
              DECIMAL(5, 1)),
      },
      make_flat_vector<bool>(
          70, [](auto row) { return row % 2 == 0 ? false : true; }));

  // 90% rows selected.
  SelectivityVector row(130);
  row.setValidRange(0, 13, false);
  testCompareExpr(
      "decimal_greaterthan(c0, c1)",
      {
          make_flat_vector<int64_t>(
              130,
              [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
              nullptr,
              DECIMAL(6, 2)),
          make_flat_vector<int64_t>(
              130,
              [](auto row) { return row % 2 == 0 ? 100 : 130; },
              nullptr,
              DECIMAL(5, 1)),
      },
      make_flat_vector<bool>(
          130, [](auto row) { return row % 2 == 0 ? false : true; }),
      row);

  testCompareExpr(
      "decimal_greaterthan(c0, c1)",
      {
          wrap_in_dictionary(
              make_indices(140, [](auto row) { return row % 2; }),
              make_flat_vector<int64_t>(
                  2,
                  [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
                  nullptr,
                  DECIMAL(6, 2))),
          make_flat_vector<int64_t>(
              140,
              [](auto row) { return row % 2 == 0 ? 100 : 130; },
              nullptr,
              DECIMAL(5, 1)),
      },
      make_flat_vector<bool>(
          140, [](auto row) { return row % 2 == 0 ? false : true; }));
}

TEST_F(DecimalCompareTest, gte) {
  // Fast path when c1 vector is constant.
  testCompareExpr(
      "decimal_greaterthanorequal(c0, c1)",
      {
          make_flat_vector<int64_t>({1000, 2000, 3000, 400}, DECIMAL(6, 2)),
          make_constant((int64_t)100, 4, DECIMAL(5, 1)),
      },
      make_flat_vector<bool>({true, true, true, false}));
  testCompareExpr(
      "decimal_greaterthanorequal(c0, c1)",
      {
          make_flat_vector<int128_t>(
              {10'000'000'000'000'000,
               12'000'000'000'000'000,
               13'000'000'000'000'000,
               35'000'000'000'000'000},
              DECIMAL(38, 34)),
          make_constant((int64_t)100000, 4, DECIMAL(6, 1)),
      },
      make_flat_vector<bool>({false, false, false, false}));

  // Fast path when vectors are flat.
  testCompareExpr(
      "decimal_greaterthanorequal(c0, c1)",
      {
          make_flat_vector<int64_t>({1000, 2000, 3000, 400}, DECIMAL(6, 2)),
          make_flat_vector<int64_t>({100, 120, 130, 350}, DECIMAL(5, 1)),
      },
      make_flat_vector<bool>({true, true, true, false}));
  testCompareExpr(
      "decimal_greaterthanorequal(c0, c1)",
      {
          make_flat_vector<int128_t>(
              {10'000'000'000'000'000,
               12'000'000'000'000'000,
               13'000'000'000'000'000,
               35'000'000'000'000'000},
              DECIMAL(38, 34)),
          make_flat_vector<int64_t>(
              {100000, 120000, 130000, 350000}, DECIMAL(6, 1)),
      },
      make_flat_vector<bool>({false, false, false, false}));

  // General case when vectors are dictionary-encoded.
  testCompareExpr(
      "decimal_greaterthanorequal(c0, c1)",
      {
          wrap_in_dictionary(
              make_indices({0, 1, 2, 3}),
              make_flat_vector<int64_t>({1000, 2000, 3000, 400}, DECIMAL(6, 2))),
          wrap_in_dictionary(
              make_indices({0, 1, 2, 3}),
              make_flat_vector<int64_t>({100, 120, 130, 350}, DECIMAL(5, 1))),
      },
      make_flat_vector<bool>({true, true, true, false}));
  testCompareExpr(
      "decimal_greaterthanorequal(c0, c1)",
      {
          wrap_in_dictionary(
              make_indices({0, 1, 2, 3}),
              make_flat_vector<int128_t>(
                  {10'000'000'000'000'000,
                   12'000'000'000'000'000,
                   13'000'000'000'000'000,
                   35'000'000'000'000'000},
                  DECIMAL(38, 34))),
          wrap_in_dictionary(
              make_indices({0, 1, 2, 3}),
              make_flat_vector<int64_t>(
                  {100000, 120000, 130000, 350000}, DECIMAL(6, 1))),
      },
      make_flat_vector<bool>({false, false, false, false}));

  // Decimal with nulls.
  testCompareExpr(
      "decimal_greaterthanorequal(c0, c1)",
      {
          make_flat_vector<int64_t>({1000, 2000, 3000, 400}, DECIMAL(6, 2)),
          make_nullable_flat_vector<int64_t>(
              {100, std::nullopt, 130, 350}, DECIMAL(5, 1)),
      },
      make_nullable_flat_vector<bool>({true, std::nullopt, true, false}));
  testCompareExpr(
      "decimal_greaterthanorequal(c0, c1)",
      {
          make_flat_vector<int128_t>(
              {10'000'000'000'000'000,
               12'000'000'000'000'000,
               13'000'000'000'000'000,
               35'000'000'000'000'000},
              DECIMAL(38, 34)),
          make_nullable_flat_vector<int64_t>(
              {100000, std::nullopt, 130000, 350000}, DECIMAL(6, 1)),
      },
      make_nullable_flat_vector<bool>({false, std::nullopt, false, false}));

  // All rows selected.
  testCompareExpr(
      "decimal_greaterthanorequal(c0, c1)",
      {
          make_flat_vector<int64_t>(
              70,
              [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
              nullptr,
              DECIMAL(6, 2)),
          make_flat_vector<int64_t>(
              70,
              [](auto row) { return row % 2 == 0 ? 100 : 130; },
              nullptr,
              DECIMAL(5, 1)),
      },
      make_flat_vector<bool>(70, [](auto /*row*/) { return true; }));

  // 90% rows selected.
  SelectivityVector row(130);
  row.setValidRange(0, 13, false);
  testCompareExpr(
      "decimal_greaterthanorequal(c0, c1)",
      {
          make_flat_vector<int64_t>(
              130,
              [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
              nullptr,
              DECIMAL(6, 2)),
          make_flat_vector<int64_t>(
              130,
              [](auto row) { return row % 2 == 0 ? 100 : 130; },
              nullptr,
              DECIMAL(5, 1)),
      },
      make_flat_vector<bool>(130, [](auto /*row*/) { return true; }),
      row);

  testCompareExpr(
      "decimal_greaterthanorequal(c0, c1)",
      {
          wrap_in_dictionary(
              make_indices(140, [](auto row) { return row % 2; }),
              make_flat_vector<int64_t>(
                  2,
                  [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
                  nullptr,
                  DECIMAL(6, 2))),
          make_flat_vector<int64_t>(
              140,
              [](auto row) { return row % 2 == 0 ? 100 : 130; },
              nullptr,
              DECIMAL(5, 1)),
      },
      make_flat_vector<bool>(140, [](auto /*row*/) { return true; }));
}

TEST_F(DecimalCompareTest, eq) {
  // Fast path when c1 vector is constant.
  testCompareExpr(
      "decimal_equalto(c0, c1)",
      {
          make_flat_vector<int64_t>({1000, 2000, 3000, 400}, DECIMAL(6, 2)),
          make_constant((int64_t)100, 4, DECIMAL(5, 1)),
      },
      make_flat_vector<bool>({true, false, false, false}));
  testCompareExpr(
      "decimal_equalto(c0, c1)",
      {
          make_flat_vector<int128_t>(
              {10'000'000'000'000'000,
               12'000'000'000'000'000,
               13'000'000'000'000'000,
               35'000'000'000'000'000},
              DECIMAL(38, 34)),
          make_constant((int64_t)100000, 4, DECIMAL(6, 1)),
      },
      make_flat_vector<bool>({false, false, false, false}));

  // Fast path when vectors are flat.
  testCompareExpr(
      "decimal_equalto(c0, c1)",
      {
          make_flat_vector<int64_t>({1000, 2000, 3000, 400}, DECIMAL(6, 2)),
          make_flat_vector<int64_t>({100, 120, 130, 350}, DECIMAL(5, 1)),
      },
      make_flat_vector<bool>({true, false, false, false}));
  testCompareExpr(
      "decimal_equalto(c0, c1)",
      {
          make_flat_vector<int128_t>(
              {10'000'000'000'000'000,
               12'000'000'000'000'000,
               13'000'000'000'000'000,
               35'000'000'000'000'000},
              DECIMAL(38, 34)),
          make_flat_vector<int64_t>(
              {100000, 120000, 130000, 350000}, DECIMAL(6, 1)),
      },
      make_flat_vector<bool>({false, false, false, false}));

  // General case when vectors are dictionary-encoded.
  testCompareExpr(
      "decimal_equalto(c0, c1)",
      {
          wrap_in_dictionary(
              make_indices({0, 1, 2, 3}),
              make_flat_vector<int64_t>({1000, 2000, 3000, 400}, DECIMAL(6, 2))),
          wrap_in_dictionary(
              make_indices({0, 1, 2, 3}),
              make_flat_vector<int64_t>({100, 120, 130, 350}, DECIMAL(5, 1))),
      },
      make_flat_vector<bool>({true, false, false, false}));
  testCompareExpr(
      "decimal_equalto(c0, c1)",
      {
          wrap_in_dictionary(
              make_indices({0, 1, 2, 3}),
              make_flat_vector<int128_t>(
                  {10'000'000'000'000'000,
                   12'000'000'000'000'000,
                   13'000'000'000'000'000,
                   35'000'000'000'000'000},
                  DECIMAL(38, 34))),
          wrap_in_dictionary(
              make_indices({0, 1, 2, 3}),
              make_flat_vector<int64_t>(
                  {100000, 120000, 130000, 350000}, DECIMAL(6, 1))),
      },
      make_flat_vector<bool>({false, false, false, false}));

  // Decimal with nulls.
  testCompareExpr(
      "decimal_equalto(c0, c1)",
      {
          make_flat_vector<int64_t>({1000, 2000, 3000, 400}, DECIMAL(6, 2)),
          make_nullable_flat_vector<int64_t>(
              {100, std::nullopt, 130, 350}, DECIMAL(5, 1)),
      },
      make_nullable_flat_vector<bool>({true, std::nullopt, false, false}));
  testCompareExpr(
      "decimal_equalto(c0, c1)",
      {
          make_flat_vector<int128_t>(
              {10'000'000'000'000'000,
               12'000'000'000'000'000,
               13'000'000'000'000'000,
               35'000'000'000'000'000},
              DECIMAL(38, 34)),
          make_nullable_flat_vector<int64_t>(
              {100000, std::nullopt, 130000, 350000}, DECIMAL(6, 1)),
      },
      make_nullable_flat_vector<bool>({false, std::nullopt, false, false}));

  // All rows selected.
  testCompareExpr(
      "decimal_equalto(c0, c1)",
      {
          make_flat_vector<int64_t>(
              70,
              [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
              nullptr,
              DECIMAL(6, 2)),
          make_flat_vector<int64_t>(
              70,
              [](auto row) { return row % 2 == 0 ? 100 : 130; },
              nullptr,
              DECIMAL(5, 1)),
      },
      make_flat_vector<bool>(
          70, [](auto row) { return row % 2 == 0 ? true : false; }));

  // 90% rows selected.
  SelectivityVector row(130);
  row.setValidRange(0, 13, false);
  testCompareExpr(
      "decimal_equalto(c0, c1)",
      {
          make_flat_vector<int64_t>(
              130,
              [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
              nullptr,
              DECIMAL(6, 2)),
          make_flat_vector<int64_t>(
              130,
              [](auto row) { return row % 2 == 0 ? 100 : 130; },
              nullptr,
              DECIMAL(5, 1)),
      },
      make_flat_vector<bool>(
          130, [](auto row) { return row % 2 == 0 ? true : false; }),
      row);

  testCompareExpr(
      "decimal_equalto(c0, c1)",
      {
          make_flat_vector<int64_t>(
              140,
              [](auto row) { return row % 2 == 0 ? 100 : 130; },
              nullptr,
              DECIMAL(5, 1)),
          wrap_in_dictionary(
              make_indices(140, [](auto row) { return row % 2; }),
              make_flat_vector<int64_t>(
                  2,
                  [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
                  nullptr,
                  DECIMAL(6, 2))),
      },
      make_flat_vector<bool>(
          140, [](auto row) { return row % 2 == 0 ? true : false; }));
}

TEST_F(DecimalCompareTest, neq) {
  // Fast path when c1 vector is constant.
  testCompareExpr(
      "decimal_notequalto(c0, c1)",
      {
          make_flat_vector<int64_t>({1000, 2000, 3000, 400}, DECIMAL(6, 2)),
          make_constant((int64_t)100, 4, DECIMAL(5, 1)),
      },
      make_flat_vector<bool>({false, true, true, true}));
  testCompareExpr(
      "decimal_notequalto(c0, c1)",
      {
          make_flat_vector<int128_t>(
              {10'000'000'000'000'000,
               12'000'000'000'000'000,
               13'000'000'000'000'000,
               35'000'000'000'000'000},
              DECIMAL(38, 34)),
          make_constant((int64_t)100000, 4, DECIMAL(6, 1)),
      },
      make_flat_vector<bool>({true, true, true, true}));

  // Fast path when vectors are flat.
  testCompareExpr(
      "decimal_notequalto(c0, c1)",
      {
          make_flat_vector<int64_t>({1000, 2000, 3000, 400}, DECIMAL(6, 2)),
          make_flat_vector<int64_t>({100, 120, 130, 350}, DECIMAL(5, 1)),
      },
      make_flat_vector<bool>({false, true, true, true}));
  testCompareExpr(
      "decimal_notequalto(c0, c1)",
      {
          make_flat_vector<int128_t>(
              {10'000'000'000'000'000,
               12'000'000'000'000'000,
               13'000'000'000'000'000,
               35'000'000'000'000'000},
              DECIMAL(38, 34)),
          make_flat_vector<int64_t>(
              {100000, 120000, 130000, 350000}, DECIMAL(6, 1)),
      },
      make_flat_vector<bool>({true, true, true, true}));

  // General case when vectors are dictionary-encoded.
  testCompareExpr(
      "decimal_notequalto(c0, c1)",
      {
          wrap_in_dictionary(
              make_indices({0, 1, 2, 3}),
              make_flat_vector<int64_t>({1000, 2000, 3000, 400}, DECIMAL(6, 2))),
          wrap_in_dictionary(
              make_indices({0, 1, 2, 3}),
              make_flat_vector<int64_t>({100, 120, 130, 350}, DECIMAL(5, 1))),
      },
      make_flat_vector<bool>({false, true, true, true}));
  testCompareExpr(
      "decimal_notequalto(c0, c1)",
      {
          wrap_in_dictionary(
              make_indices({0, 1, 2, 3}),
              make_flat_vector<int128_t>(
                  {10'000'000'000'000'000,
                   12'000'000'000'000'000,
                   13'000'000'000'000'000,
                   35'000'000'000'000'000},
                  DECIMAL(38, 34))),
          wrap_in_dictionary(
              make_indices({0, 1, 2, 3}),
              make_flat_vector<int64_t>(
                  {100000, 120000, 130000, 350000}, DECIMAL(6, 1))),
      },
      make_flat_vector<bool>({true, true, true, true}));

  // Decimal with nulls.
  testCompareExpr(
      "decimal_notequalto(c0, c1)",
      {
          make_flat_vector<int64_t>({1000, 2000, 3000, 400}, DECIMAL(6, 2)),
          make_nullable_flat_vector<int64_t>(
              {100, std::nullopt, 130, 350}, DECIMAL(5, 1)),
      },
      make_nullable_flat_vector<bool>({false, std::nullopt, true, true}));
  testCompareExpr(
      "decimal_notequalto(c0, c1)",
      {
          make_flat_vector<int128_t>(
              {10'000'000'000'000'000,
               12'000'000'000'000'000,
               13'000'000'000'000'000,
               35'000'000'000'000'000},
              DECIMAL(38, 34)),
          make_nullable_flat_vector<int64_t>(
              {100000, std::nullopt, 130000, 350000}, DECIMAL(6, 1)),
      },
      make_nullable_flat_vector<bool>({true, std::nullopt, true, true}));

  // All rows selected.
  testCompareExpr(
      "decimal_notequalto(c0, c1)",
      {
          make_flat_vector<int64_t>(
              70,
              [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
              nullptr,
              DECIMAL(6, 2)),
          make_flat_vector<int64_t>(
              70,
              [](auto row) { return row % 2 == 0 ? 100 : 130; },
              nullptr,
              DECIMAL(5, 1)),
      },
      make_flat_vector<bool>(
          70, [](auto row) { return row % 2 == 0 ? false : true; }));

  // 90% rows selected.
  SelectivityVector row(130);
  row.setValidRange(0, 13, false);
  testCompareExpr(
      "decimal_notequalto(c0, c1)",
      {
          make_flat_vector<int64_t>(
              130,
              [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
              nullptr,
              DECIMAL(6, 2)),
          make_flat_vector<int64_t>(
              130,
              [](auto row) { return row % 2 == 0 ? 100 : 130; },
              nullptr,
              DECIMAL(5, 1)),
      },
      make_flat_vector<bool>(
          130, [](auto row) { return row % 2 == 0 ? false : true; }),
      row);

  testCompareExpr(
      "decimal_notequalto(c0, c1)",
      {
          wrap_in_dictionary(
              make_indices(140, [](auto row) { return row % 2; }),
              make_flat_vector<int64_t>(
                  2,
                  [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
                  nullptr,
                  DECIMAL(6, 2))),
          make_flat_vector<int64_t>(
              140,
              [](auto row) { return row % 2 == 0 ? 100 : 130; },
              nullptr,
              DECIMAL(5, 1)),
      },
      make_flat_vector<bool>(
          140, [](auto row) { return row % 2 == 0 ? false : true; }));
}

TEST_F(DecimalCompareTest, lt) {
  // Fast path when c1 vector is constant.
  testCompareExpr(
      "decimal_lessthan(c0, c1)",
      {
          make_flat_vector<int64_t>({1000, 2000, 3000, 400}, DECIMAL(6, 2)),
          make_constant((int64_t)100, 4, DECIMAL(5, 1)),
      },
      make_flat_vector<bool>({false, false, false, true}));
  testCompareExpr(
      "decimal_lessthan(c0, c1)",
      {
          make_flat_vector<int128_t>(
              {10'000'000'000'000'000,
               12'000'000'000'000'000,
               13'000'000'000'000'000,
               35'000'000'000'000'000},
              DECIMAL(38, 34)),
          make_constant((int64_t)100000, 4, DECIMAL(6, 1)),
      },
      make_flat_vector<bool>({true, true, true, true}));

  // Fast path when vectors are flat.
  testCompareExpr(
      "decimal_lessthan(c0, c1)",
      {
          make_flat_vector<int64_t>({1000, 2000, 3000, 400}, DECIMAL(6, 2)),
          make_flat_vector<int64_t>({100, 120, 130, 350}, DECIMAL(5, 1)),
      },
      make_flat_vector<bool>({false, false, false, true}));
  testCompareExpr(
      "decimal_lessthan(c0, c1)",
      {
          make_flat_vector<int128_t>(
              {10'000'000'000'000'000,
               12'000'000'000'000'000,
               13'000'000'000'000'000,
               35'000'000'000'000'000},
              DECIMAL(38, 34)),
          make_flat_vector<int64_t>(
              {100000, 120000, 130000, 350000}, DECIMAL(6, 1)),
      },
      make_flat_vector<bool>({true, true, true, true}));

  // General case when vectors are dictionary-encoded.
  testCompareExpr(
      "decimal_lessthan(c0, c1)",
      {
          wrap_in_dictionary(
              make_indices({0, 1, 2, 3}),
              make_flat_vector<int64_t>({1000, 2000, 3000, 400}, DECIMAL(6, 2))),
          wrap_in_dictionary(
              make_indices({0, 1, 2, 3}),
              make_flat_vector<int64_t>({100, 120, 130, 350}, DECIMAL(5, 1))),
      },
      make_flat_vector<bool>({false, false, false, true}));
  testCompareExpr(
      "decimal_lessthan(c0, c1)",
      {
          wrap_in_dictionary(
              make_indices({0, 1, 2, 3}),
              make_flat_vector<int128_t>(
                  {10'000'000'000'000'000,
                   12'000'000'000'000'000,
                   13'000'000'000'000'000,
                   35'000'000'000'000'000},
                  DECIMAL(38, 34))),
          wrap_in_dictionary(
              make_indices({0, 1, 2, 3}),
              make_flat_vector<int64_t>(
                  {100000, 120000, 130000, 350000}, DECIMAL(6, 1))),
      },
      make_flat_vector<bool>({true, true, true, true}));

  // Decimal with nulls.
  testCompareExpr(
      "decimal_lessthan(c0, c1)",
      {
          make_flat_vector<int64_t>({1000, 2000, 3000, 400}, DECIMAL(6, 2)),
          make_nullable_flat_vector<int64_t>(
              {100, std::nullopt, 130, 350}, DECIMAL(5, 1)),
      },
      make_nullable_flat_vector<bool>({false, std::nullopt, false, true}));
  testCompareExpr(
      "decimal_lessthan(c0, c1)",
      {
          make_flat_vector<int128_t>(
              {10'000'000'000'000'000,
               12'000'000'000'000'000,
               13'000'000'000'000'000,
               35'000'000'000'000'000},
              DECIMAL(38, 34)),
          make_nullable_flat_vector<int64_t>(
              {100000, std::nullopt, 130000, 350000}, DECIMAL(6, 1)),
      },
      make_nullable_flat_vector<bool>({true, std::nullopt, true, true}));

  // All rows selected.
  testCompareExpr(
      "decimal_lessthan(c0, c1)",
      {
          make_flat_vector<int64_t>(
              70,
              [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
              nullptr,
              DECIMAL(6, 2)),
          make_flat_vector<int64_t>(
              70,
              [](auto row) { return row % 2 == 0 ? 100 : 130; },
              nullptr,
              DECIMAL(5, 1)),
      },
      make_flat_vector<bool>(70, [](auto /*row*/) { return false; }));

  // 90% rows selected.
  SelectivityVector row(130);
  row.setValidRange(0, 13, false);
  testCompareExpr(
      "decimal_lessthan(c0, c1)",
      {
          make_flat_vector<int64_t>(
              130,
              [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
              nullptr,
              DECIMAL(6, 2)),
          make_flat_vector<int64_t>(
              130,
              [](auto row) { return row % 2 == 0 ? 100 : 130; },
              nullptr,
              DECIMAL(5, 1)),
      },
      make_flat_vector<bool>(130, [](auto /*row*/) { return false; }),
      row);

  testCompareExpr(
      "decimal_lessthan(c0, c1)",
      {
          wrap_in_dictionary(
              make_indices(140, [](auto row) { return row % 2; }),
              make_flat_vector<int64_t>(
                  2,
                  [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
                  nullptr,
                  DECIMAL(6, 2))),
          make_flat_vector<int64_t>(
              140,
              [](auto row) { return row % 2 == 0 ? 100 : 130; },
              nullptr,
              DECIMAL(5, 1)),
      },
      make_flat_vector<bool>(140, [](auto /*row*/) { return false; }));
}

TEST_F(DecimalCompareTest, lte) {
  // Fast path when c1 vector is constant.
  testCompareExpr(
      "decimal_lessthanorequal(c0, c1)",
      {
          make_flat_vector<int64_t>({1000, 2000, 3000, 400}, DECIMAL(6, 2)),
          make_constant((int64_t)100, 4, DECIMAL(5, 1)),
      },
      make_flat_vector<bool>({true, false, false, true}));
  testCompareExpr(
      "decimal_lessthanorequal(c0, c1)",
      {
          make_flat_vector<int128_t>(
              {10'000'000'000'000'000,
               12'000'000'000'000'000,
               13'000'000'000'000'000,
               35'000'000'000'000'000},
              DECIMAL(38, 34)),
          make_constant((int64_t)100000, 4, DECIMAL(6, 1)),
      },
      make_flat_vector<bool>({true, true, true, true}));

  // Fast path when vectors are flat.
  testCompareExpr(
      "decimal_lessthanorequal(c0, c1)",
      {
          make_flat_vector<int64_t>({1000, 2000, 3000, 400}, DECIMAL(6, 2)),
          make_flat_vector<int64_t>({100, 120, 130, 350}, DECIMAL(5, 1)),
      },
      make_flat_vector<bool>({true, false, false, true}));
  testCompareExpr(
      "decimal_lessthanorequal(c0, c1)",
      {
          make_flat_vector<int128_t>(
              {10'000'000'000'000'000,
               12'000'000'000'000'000,
               13'000'000'000'000'000,
               35'000'000'000'000'000},
              DECIMAL(38, 34)),
          make_flat_vector<int64_t>(
              {100000, 120000, 130000, 350000}, DECIMAL(6, 1)),
      },
      make_flat_vector<bool>({true, true, true, true}));

  // General case when vectors are dictionary-encoded.
  testCompareExpr(
      "decimal_lessthanorequal(c0, c1)",
      {
          wrap_in_dictionary(
              make_indices({0, 1, 2, 3}),
              make_flat_vector<int64_t>({1000, 2000, 3000, 400}, DECIMAL(6, 2))),
          wrap_in_dictionary(
              make_indices({0, 1, 2, 3}),
              make_flat_vector<int64_t>({100, 120, 130, 350}, DECIMAL(5, 1))),
      },
      make_flat_vector<bool>({true, false, false, true}));
  testCompareExpr(
      "decimal_lessthanorequal(c0, c1)",
      {
          wrap_in_dictionary(
              make_indices({0, 1, 2, 3}),
              make_flat_vector<int128_t>(
                  {10'000'000'000'000'000,
                   12'000'000'000'000'000,
                   13'000'000'000'000'000,
                   35'000'000'000'000'000},
                  DECIMAL(38, 34))),
          wrap_in_dictionary(
              make_indices({0, 1, 2, 3}),
              make_flat_vector<int64_t>(
                  {100000, 120000, 130000, 350000}, DECIMAL(6, 1))),
      },
      make_flat_vector<bool>({true, true, true, true}));

  // Decimal with nulls.
  testCompareExpr(
      "decimal_lessthanorequal(c0, c1)",
      {
          make_flat_vector<int64_t>({1000, 2000, 3000, 400}, DECIMAL(6, 2)),
          make_nullable_flat_vector<int64_t>(
              {100, std::nullopt, 130, 350}, DECIMAL(5, 1)),
      },
      make_nullable_flat_vector<bool>({true, std::nullopt, false, true}));
  testCompareExpr(
      "decimal_lessthanorequal(c0, c1)",
      {
          make_flat_vector<int128_t>(
              {10'000'000'000'000'000,
               12'000'000'000'000'000,
               13'000'000'000'000'000,
               35'000'000'000'000'000},
              DECIMAL(38, 34)),
          make_nullable_flat_vector<int64_t>(
              {100000, std::nullopt, 130000, 350000}, DECIMAL(6, 1)),
      },
      make_nullable_flat_vector<bool>({true, std::nullopt, true, true}));

  // All rows selected.
  testCompareExpr(
      "decimal_lessthanorequal(c0, c1)",
      {
          make_flat_vector<int64_t>(
              70,
              [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
              nullptr,
              DECIMAL(6, 2)),
          make_flat_vector<int64_t>(
              70,
              [](auto row) { return row % 2 == 0 ? 100 : 130; },
              nullptr,
              DECIMAL(5, 1)),
      },
      make_flat_vector<bool>(
          70, [](auto row) { return row % 2 == 0 ? true : false; }));

  // 90% rows selected.
  SelectivityVector row(130);
  row.setValidRange(0, 13, false);
  testCompareExpr(
      "decimal_lessthanorequal(c0, c1)",
      {
          make_flat_vector<int64_t>(
              130,
              [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
              nullptr,
              DECIMAL(6, 2)),
          make_flat_vector<int64_t>(
              130,
              [](auto row) { return row % 2 == 0 ? 100 : 130; },
              nullptr,
              DECIMAL(5, 1)),
      },
      make_flat_vector<bool>(
          130, [](auto row) { return row % 2 == 0 ? true : false; }),
      row);

  testCompareExpr(
      "decimal_lessthanorequal(c0, c1)",
      {
          wrap_in_dictionary(
              make_indices(140, [](auto row) { return row % 2; }),
              make_flat_vector<int64_t>(
                  2,
                  [](auto row) { return row % 2 == 0 ? 1000 : 3000; },
                  nullptr,
                  DECIMAL(6, 2))),
          make_flat_vector<int64_t>(
              140,
              [](auto row) { return row % 2 == 0 ? 100 : 130; },
              nullptr,
              DECIMAL(5, 1)),
      },
      make_flat_vector<bool>(
          140, [](auto row) { return row % 2 == 0 ? true : false; }));
}

} // namespace
} // namespace kumo::pollux::functions::sparksql::test
